package com.raymond.algorithm.currentlimit;

import com.raymond.algorithm.utils.ThreadPool;

import java.util.LinkedList;
import java.util.concurrent.TimeUnit;
import java.util.concurrent.atomic.AtomicLong;

/**
 * 滑动窗口
 * 一、回顾：计算器算法存在问题
 * 对于秒级以上的时间周期来说，会存在一个非常严重的问题，那就是临界问题。
 * 假设有一个恶意用户，他在0:59时，瞬间发送了100个请求，并且1:00又瞬间发送了100个请求，
 * 那么其实这个用户在 1秒里面，瞬间发送了200个请求。我们刚才规定的是1分钟最多100个请求，
 * 也就是每秒钟最多1.7个请求，用户通过在时间窗口的重置节点处突发请求， 可以瞬间超过我们的速率限制。
 * 用户有可能通过算法的这个漏洞，瞬间压垮我们的应用
 *
 * 二.问题点
 * 计算器的问题其实是因为我们统计的精度太低（统计的1秒内的流量情况，只有1个值进行记录）。
 * 那么如何很好地处理这个问题呢？或者说，如何将临界问题的影响降低呢？我们可以看下面的滑动窗口算法。
 *
 * 三.原理
 * 1.滑动窗口，又称rolling window。为了解决计数器法统计精度太低的问题，引入了滑动窗口算法。
 * 2.是一分钟。然后我们将时间窗口进行划分。我们就将滑动窗口划成了6格，所以每格代表的是10秒钟。每过10秒钟，我们的时间窗口就会往右滑动一格。
 * 每一个格子都有自己独立的计数器counter，比如当一个请求 在0:35秒的时候到达，那么0:30~0:39对应的counter就会加1。
 * 3.那么滑动窗口怎么解决刚才的临界问题的呢？
 * 0:59到达的100个请求会落在灰色的格子中，而1:00到达的请求会落在橘×××的格子中。
 * 当时间到达1:00时，我们的窗口会往右移动一格，那么此时时间窗口内的总请求数量一共是200个，超过了限定的100个，所以此时能够检测出来触发了限流。
 * 4.由此可见，当滑动窗口的格子划分的越多，那么滑动窗口的滚动就越平滑，限流的统计就会越精确。
 * （1）滑动窗口算法是以当前这个时间点为基准，往前推移1秒进行计算当前1秒内的请求量情况。
 * （2）滑动窗口限流统计的精准度是由划分的格子多少决定的，这个怎么理解呐，就是把1秒中进行划分成多个时间段，比如2个格子的话，那么就是2段，0-500ms和501-1000ms。那么就会两个值进行存储统计请求量，比如数组[0,1] 各存储一个段的请求值。
 * （3）计算器算法是滑动窗口算法将时间段划分为1的特殊情况。
 *
 * @author :  raymond
 * @version :  V1.0
 * @date :  2021-04-22 15:31
 */
public class SlidingTimeWindow {
    /**
     * 线程滑动窗口
     */
    public static final int THREAD = 0;
    /**
     * 环形滑动窗口
     */
    public static final int ANNULUS = 1;

    private AbstractWindow abstractWindow;

    public SlidingTimeWindow(int limit, int split) {
        this(limit, split, THREAD);
    }

    public SlidingTimeWindow(int limit, int split, int type) {
        if (THREAD == type) {
            abstractWindow = new ThreadSlidingWindow(limit, split);
        }
        if (ANNULUS == type) {
            abstractWindow = new AnnulusSlidingWindow(limit, split);
        }
    }

    public boolean limit() {
        return abstractWindow.limit();
    }

    /**
     * 抽象窗口
     */
    private abstract class AbstractWindow {
        /**
         * 限流
         * true 限流了
         * false 允许正常访问
         * @return true,false
         */
        abstract boolean limit();
    }

    /**
     * 通过定时线程滑动窗口
     *
     */
    private class ThreadSlidingWindow extends AbstractWindow{
        /**
         * 时间窗口最大请求数
         */
        private final int limit;
        /**
         * 服务访问次数
         */
        private AtomicLong counter = new AtomicLong();
        /**
         * 使用List来记录滑动窗口的10个格子。
         */
        private LinkedList<Long> slots = new LinkedList<Long>();
        /**
         * 时间划分段落
         */
        private final int split;


        public ThreadSlidingWindow(int limit, int split) {
            this.limit = limit;
            this.split = split;
            slots.add(0L);
            ThreadPool.javaScheduledThreadExecutor("滑动窗口线程")
                    .scheduleWithFixedDelay(() -> {
                        slots.add(counter.get());
                        if (slots.size() > split) {
                            slots.removeFirst();
                        }
                    }, 0, 1000 / split, TimeUnit.MILLISECONDS);
        }

        @Override
        public boolean limit() {
            return counter.incrementAndGet() - slots.getFirst() > limit;
        }
    }

    /**
     * 通过环形数组滑动窗口
     */
    private class AnnulusSlidingWindow extends AbstractWindow {

        /**
         * 时间窗口最大请求数
         */
        private final int limit;

        /**
         * 时间划分段落
         */
        private final int split;

        /**
         * 整个大窗口的时间
         */
        private final long windowTime = 1000;

        /**
         * 窗口
         */
        private WindowWrap[] slots;
        /**
         * 窗口大小
         */
        private long windowLengthInMs;

        private AnnulusSlidingWindow(int limit, int split) {
            this.limit = limit;
            this.split = split;
            this.slots = new WindowWrap[split];
            for (int i = 0; i < split; i++) {
                this.slots[i] = new WindowWrap();
            }
            this.windowLengthInMs = 1000 / split;
        }

        @Override
        boolean limit() {
            long now = System.currentTimeMillis();
            WindowWrap slot = getWindowWrap(now);
            long start = calculateWindowStart(now);
            addWaitingRequest(slot, start);
            return isLimit(slot);
        }

        private WindowWrap getWindowWrap(long now) {
            int index = index(now);
            return slots[index];
        }

        private boolean isLimit(WindowWrap slot) {
            synchronized (slot.object) {
                int count = 0;
                for (WindowWrap windowWrap : slots) {
                    count += windowWrap.count.get();
                }
                if (count < limit) {
                    slot.count.incrementAndGet();
                    return false;
                }
            }
            return true;
        }

        private void addWaitingRequest(WindowWrap slot, long start) {
            if (start > slot.time) {
                resetWindowTo(start);
                slot.time = start;
            }
        }

        private void resetWindowTo(long start) {
            for (WindowWrap slot : slots) {
                if (start - slot.time >= windowTime) {
                    slot.count.set(0);
                }
            }
        }

        /**
         * 获取小窗口的时间
         * @param timeMillis 时间戳
         * @return 小窗口的时间
         */
        private long calculateWindowStart(long timeMillis) {
            return timeMillis - timeMillis % windowLengthInMs;
        }

        /**
         * 获取小窗口的下标
         * @param now 时间戳
         * @return 小窗口的时间
         */
        private int index(long now) {
            long timeId = now / windowLengthInMs;
            return (int)(timeId % slots.length);
        }

        private class WindowWrap {
            /**
             * 时间
             */
            private long time;
            /**
             * 请求次数
             */
            private AtomicLong count = new AtomicLong();

            private final Object object = new Object();
        }
    }


    public static void main(String[] args) throws InterruptedException {
        SlidingTimeWindow slidingTimeWindow = new SlidingTimeWindow(2000, 1000, 1);
        long start = System.currentTimeMillis();
        for (int i = 0; i < 5000; i++) {
            boolean limit = slidingTimeWindow.limit();
            if (limit) {
//                System.out.println("请求" + i);
//                System.out.println("当前时间:" + (System.currentTimeMillis() - start));
                TimeUnit.MILLISECONDS.sleep(100);
                i --;
            }
            if (i == 0) {
//                TimeUnit.SECONDS.sleep(9);
            }
        }
        System.out.println(System.currentTimeMillis() - start);
    }


}
